package com.gph.saviorframework.mq.service.impl;

import com.alibaba.rocketmq.common.TopicConfig;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.admin.TopicOffset;
import com.alibaba.rocketmq.common.admin.TopicStatsTable;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.body.TopicList;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
import com.alibaba.rocketmq.tools.command.CommandUtil;
import com.alibaba.rocketmq.tools.command.topic.DeleteTopicSubCommand;
import com.alibaba.rocketmq.tools.command.topic.TopicListSubCommand;
import com.alibaba.rocketmq.tools.command.topic.TopicStatusSubCommand;
import com.alibaba.rocketmq.tools.command.topic.UpdateTopicSubCommand;
import com.gph.saviorframework.mq.service.TopicService;
import com.gph.saviorframework.mq.service.impl.validate.CmdTrace;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.cli.Option;

import java.util.*;

/**
 * topic service impl
 */
public class TopicServiceImpl extends AbstractService implements TopicService {

    /**
     * get all topics
     */
    @CmdTrace(cmdClazz = TopicListSubCommand.class)
    public List<String> getTopics(){

        List<String> topics = new ArrayList<String>();

        DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
        try {
            defaultMQAdminExt.start();
            TopicList topicList = defaultMQAdminExt.fetchAllTopicList();
            if(topicList!=null){
                for(String topicName : topicList.getTopicList()){
                    topics.add(topicName);
                }
            }
        }
        catch (Exception ex){
            ex.printStackTrace();
        }
        finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
        return topics;
    }

    @CmdTrace(cmdClazz = TopicListSubCommand.class)
    public TopicList getTopicList() {
        DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
        try {
            defaultMQAdminExt.start();
            TopicList topicList = defaultMQAdminExt.fetchAllTopicList();
            if(topicList!=null){
                return topicList;
            }
        }
        catch (Exception ex){
            ex.printStackTrace();
        }
        finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
        return null;
    }

    @CmdTrace(cmdClazz = UpdateTopicSubCommand.class)
    public boolean update(String topic, String readQueueNums, String writeQueueNums,
                          String perm, String brokerAddr, String clusterName) {
        DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();

        try {
            TopicConfig topicConfig = new TopicConfig();
            topicConfig.setReadQueueNums(8);
            topicConfig.setWriteQueueNums(8);
            topicConfig.setTopicName(topic);

            if (StringUtils.isNotBlank(readQueueNums)) {
                topicConfig.setReadQueueNums(Integer.parseInt(readQueueNums));
            }

            if (StringUtils.isNotBlank(writeQueueNums)) {
                topicConfig.setWriteQueueNums(Integer.parseInt(writeQueueNums));
            }

            if (StringUtils.isNotBlank(perm)) {
                topicConfig.setPerm(translatePerm(perm));
            }

            if (StringUtils.isNotBlank(brokerAddr)) {
                defaultMQAdminExt.start();
                defaultMQAdminExt.createAndUpdateTopicConfig(brokerAddr, topicConfig);
                return true;
            }
            else if (StringUtils.isNotBlank(clusterName)) {

                defaultMQAdminExt.start();

                Set<String> masterSet =
                        CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, clusterName);
                for (String addr : masterSet) {
                    defaultMQAdminExt.createAndUpdateTopicConfig(addr, topicConfig);
                }
                return true;
            }
            else {
                throw new IllegalStateException("clusterName or brokerAddr can not be all blank");
            }
        }
        catch (Throwable e) {
            e.printStackTrace();
            return false;
        }
        finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
    }

    /**
     * 删除topic
     *
     * @param topicName
     * @param clusterName
     * @return
     */
    @CmdTrace(cmdClazz = DeleteTopicSubCommand.class)
    public boolean delete(String topicName, String clusterName) {
        DefaultMQAdminExt adminExt = getDefaultMQAdminExt();
        try {
            if (StringUtils.isNotBlank(clusterName)) {
                adminExt.start();
                Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(adminExt, clusterName);
                adminExt.deleteTopicInBroker(masterSet, topicName);
                Set<String> nameServerSet = null;
                if (StringUtils.isNotBlank(configureInitializer.getNamesrvAddr())) {
                    String[] ns = configureInitializer.getNamesrvAddr().split(";");
                    nameServerSet = new HashSet<String>(Arrays.asList(ns));
                }
                adminExt.deleteTopicInNameServer(nameServerSet, topicName);
                return true;
            }
            else {
                throw new IllegalStateException("clusterName is blank");
            }
        }
        catch (Throwable e) {
            e.printStackTrace();
        }
        finally {
            shutdownDefaultMQAdminExt(adminExt);
        }
        return false;
    }

    /**
     * 获取topic状态
     *
     * @param topicName
     * @return
     */
    @CmdTrace(cmdClazz = TopicStatusSubCommand.class)
    public TopicStatsTable getTopicStats(String topicName) {
        DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
        try {
            TopicStatsTable stats = defaultMQAdminExt.examineTopicStats(topicName);
            return stats;
        }
        catch (Exception ex){
            ex.printStackTrace();
        }
        finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
        return null;
    }

    /**
     * 获取topic状态的map结果集合，供controller使用
     *
     * @param topicName
     * @return
     */
    public List<Map<String, Object>> getTopicStatsMap(String topicName) {

        List<Map<String,Object>> list = new ArrayList<Map<String, Object>>();

        TopicStatsTable topicStatsTable = getTopicStats(topicName);

        if(topicStatsTable!=null){
            List<MessageQueue> mqList = new LinkedList<MessageQueue>();
            mqList.addAll(topicStatsTable.getOffsetTable().keySet());
            Collections.sort(mqList);

            for (MessageQueue mq : mqList) {

                Map<String,Object> map = new HashMap<String, Object>();

                TopicOffset topicOffset = topicStatsTable.getOffsetTable().get(mq);

                String humanTimestamp = "";
                if (topicOffset.getLastUpdateTimestamp() > 0) {
                    humanTimestamp = UtilAll.timeMillisToHumanString2(topicOffset.getLastUpdateTimestamp());
                }

                map.put("brokerName",UtilAll.frontStringAtLeast(mq.getBrokerName(), 32));
                map.put("qid",mq.getQueueId());
                map.put("minOffSet",topicOffset.getMinOffset());
                map.put("maxOffSet",topicOffset.getMaxOffset());
                map.put("lastUpdated",humanTimestamp);

                list.add(map);
            }
        }

        return list;
    }
}
